Snowflake Connector for PostgreSQL の構成手順をまとめてみる #SnowflakeDB

-- データベースを作成
CREATE DATABASE sample_db;

\c sample_db;

-- テーブル1: customers 
CREATE TABLE customers (
    customer_id SERIAL PRIMARY KEY,  
    name VARCHAR(100),               
    email VARCHAR(100),              
    age INT                          
);

-- テーブル2: products 
CREATE TABLE products (
    product_id SERIAL PRIMARY KEY,   
    name VARCHAR(100),               
    price DECIMAL(10, 2)             
);

-- テーブル3: orders
CREATE TABLE orders (
    order_id SERIAL PRIMARY KEY,     
    customer_id INT REFERENCES customers(customer_id),  
    product_id INT REFERENCES products(product_id),     
    order_date DATE                 
);

-- customers テーブルにサンプルデータを挿入
INSERT INTO customers (name, email, age) VALUES
('Alice Smith', 'alice@example.com', 30),
('Bob Johnson', 'bob@example.com', 25),
('Charlie Brown', 'charlie@example.com', 35);

-- products テーブルにサンプルデータを挿入
INSERT INTO products (name, price) VALUES
('Laptop', 999.99),
('Smartphone', 499.99),
('Tablet', 299.99);

-- orders テーブルにサンプルデータを挿入
INSERT INTO orders (customer_id, product_id, order_date) VALUES
(1, 1, '2023-10-01'),
(2, 2, '2023-10-02'),
(3, 3, '2023-10-03');

sudo mkfs -t xfs /dev/xvdb

# Dockerストレージ用
sudo mkdir -p /var/lib/docker
sudo mount /dev/xvdb /var/lib/docker

$ sudo blkid
/dev/xvda128: SEC_TYPE="msdos" UUID="A41B-D0D1" BLOCK_SIZE="512" TYPE="vfat" PARTLABEL="EFI System Partition" PARTUUID="dc8e83c2-fd31-4403-960e-484e567be164"
/dev/xvda127: PARTLABEL="BIOS Boot Partition" PARTUUID="d0854c70-9a2b-4635-a6ce-267dedbe5605"
/dev/xvda1: LABEL="/" UUID="693eea79-11af-44b1-9c1e-01aced209966" BLOCK_SIZE="4096" TYPE="xfs" PARTLABEL="Linux" PARTUUID="8eca1f17-8f97-4b7d-9b31-ba295f584293"
/dev/xvdb: UUID="3171ca58-2dd8-4626-9f10-33dc0a298ce6" BLOCK_SIZE="512" TYPE="xfs"

sudo vi /etc/fstab

UUID=3171ca58-2dd8-4626-9f10-33dc0a298ce6  /var/lib/docker  xfs  defaults,nofail  0  2

$ lsblk
NAME      MAJ:MIN RM SIZE RO TYPE MOUNTPOINTS
xvda      202:0    0   8G  0 disk
tqxvda1   202:1    0   8G  0 part /
tqxvda127 259:0    0   1M  0 part
mqxvda128 259:1    0  10M  0 part /boot/efi
xvdb      202:16   0  50G  0 disk /var/lib/docker

$ df -h
Filesystem      Size  Used Avail Use% Mounted on
devtmpfs        4.0M     0  4.0M   0% /dev
tmpfs           7.9G     0  7.9G   0% /dev/shm
tmpfs           3.2G  456K  3.2G   1% /run
/dev/xvda1      8.0G  1.6G  6.4G  20% /
tmpfs           7.9G     0  7.9G   0% /tmp
/dev/xvda128     10M  1.3M  8.7M  13% /boot/efi
tmpfs           1.6G     0  1.6G   0% /run/user/1000
/dev/xvdb        50G  389M   50G   1% /var/lib/docker

sudo dnf update
# dockerのインストール
sudo dnf install -y docker
# dockerの開始
sudo systemctl start docker
# システムの起動時に Docker も自動的に起動するように設定
sudo systemctl enable docker
# sudoなしでdockerを実行できるようにする
sudo usermod -aG docker $USER
# dockerのソケットファイルの所有グループをdockerに変更
sudo chgrp docker /var/run/docker.sock
# dockerデーモンを再起動
sudo service docker restart
# docker compose をインストール
sudo curl -L "https://github.com/docker/compose/releases/download/v2.29.7/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose

$ docker -v
Docker version 25.0.5, build 5dc9bcc

$ docker-compose -v
Docker Compose version v2.29.7

> SHOW rds.logical_replication;
 rds.logical_replication
-------------------------
 on
(1 row)

\c sample_db;
CREATE PUBLICATION mypublication;

ALTER PUBLICATION mypublication ADD TABLE customers;
ALTER PUBLICATION mypublication ADD TABLE products;
ALTER PUBLICATION mypublication ADD TABLE orders;

-- ユーザーを作成
CREATE USER sf_connector_user WITH PASSWORD '<パスワード>';

-- rds_superuserロールを割り当てる
GRANT rds_superuser TO sf_connector_user;

-- rds_replicationロールを割り当てる
GRANT rds_replication TO sf_connector_user;

--確認
> \du sf_connector_user;
                          List of roles
     Role name     | Attributes |            Member of
-------------------+------------+---------------------------------
 sf_connector_user |            | {rds_superuser,rds_replication}

mkdir -p snowflake-connector-postgres/{agent-keys,configuration}

.
`-- snowflake-connector-postgres
    |-- agent-keys
    `-- configuration

{
  "PSQLDS1": {
    "url": "jdbc:postgresql://<RDSのエンドポイント>:5432/postgres",
    "username": "sf_connector_user",
    "password": "<パスワード>",
    "publication": "mypublication",
    "ssl": false
  }
}

JAVA_OPTS=-Xmx5g

version: '1'
services:
  postgresql-agent:
    container_name: postgresql-agent
    image: snowflakedb/database-connector-agent:latest
    volumes:
      - ./agent-keys:/home/agent/.ssh
      - ./configuration/snowflake.json:/home/agent/snowflake.json
      - ./configuration/datasources.json:/home/agent/datasources.json
    env_file:
      - configuration/postgresql.conf
    mem_limit: 6g

.
`-- snowflake-connector-postgres
    |-- agent-keys
    |-- configuration
    |   |-- datasources.json
    |   `-- postgresql.conf
    `-- docker-compose.yml

USE ROLE SYSADMIN;
--イベントテーブル格納先のデータベース・スキーマを作成
CREATE DATABASE event_db;
CREATE SCHEMA event_schema;

--イベントテーブルを作成
CREATE EVENT TABLE IF NOT EXISTS event_db.event_schema.my_event_table CHANGE_TRACKING = TRUE;

--アカウントにイベントテーブルを設定
USE ROLE ACCOUNTADMIN;
ALTER ACCOUNT SET EVENT_TABLE = event_db.event_schema.my_event_table;

.
|-- agent-keys
|-- configuration
|   |-- datasources.json
|   |-- postgresql.conf
|   `-- snowflake.json
`-- docker-compose.yml

docker-compose up -d

.
|-- agent-keys
|   |-- database-connector-agent-app-private-key.p8
|   `-- database-connector-agent-app-public-key.pub
|-- configuration
|   |-- datasources.json
|   |-- postgresql.conf
|   `-- snowflake.json
`-- docker-compose.yml

USE ROLE SYSADMIN;
CREATE DATABASE DEST_DB;

USE ROLE ACCOUNTADMIN;
GRANT CREATE SCHEMA ON DATABASE DEST_DB TO APPLICATION SNOWFLAKE_CONNECTOR_FOR_POSTGRESQL;

CALL PUBLIC.ADD_DATA_SOURCE('<data_source_name>', '<dest_db>');

{
  "PSQLDS1": {
    "url": "jdbc:postgresql://<RDSのエンドポイント>:5432/postgres",
    "username": "sf_connector_user",
    "password": "<パスワード>",
    "publication": "mypublication",
    "ssl": false
  }
}

--データソースを追加
USE DATABASE SNOWFLAKE_CONNECTOR_FOR_POSTGRESQL;
CALL PUBLIC.ADD_DATA_SOURCE('PSQLDS1', 'DEST_DB');

SELECT * FROM PUBLIC.DATA_SOURCES;

CALL SNOWFLAKE_CONNECTOR_FOR_POSTGRESQL.PUBLIC.ADD_TABLES('PSQLDS1', 'public', ARRAY_CONSTRUCT('customers'));

USE ROLE ACCOUNTADMIN;
USE DATABASE SNOWFLAKE_CONNECTOR_FOR_POSTGRESQL;
SHOW GRANTS TO APPLICATION ROLE SNOWFLAKE_CONNECTOR_FOR_POSTGRESQL.DATA_READER;

GRANT USAGE ON SCHEMA  DEST_DB."public" TO ROLE SYSADMIN;
GRANT APPLICATION ROLE SNOWFLAKE_CONNECTOR_FOR_POSTGRESQL.DATA_READER TO ROLE SYSADMIN;

USE ROLE ACCOUNTADMIN;
USE DATABASE SNOWFLAKE_CONNECTOR_FOR_POSTGRESQL;
CALL SNOWFLAKE_CONNECTOR_FOR_POSTGRESQL.PUBLIC.ADD_TABLES('PSQLDS1', 'public', ARRAY_CONSTRUCT('orders','products'));

> SELECT * FROM customers;
 customer_id |     name      |        email        | age
-------------+---------------+---------------------+-----
           1 | Alice Smith   | alice@example.com   |  30
           2 | Bob Johnson   | bob@example.com     |  25
           3 | Charlie Brown | charlie@example.com |  35
(3 rows)

UPDATE customers SET age = 30 WHERE customer_id = 2;

INSERT INTO customers (name, email, age) VALUES
('David Green', 'david@example.com', 40),
('Eve White', 'eve@example.com', 28);

DELETE FROM customers WHERE customer_id = 4;

ALTER TABLE customers ADD COLUMN address VARCHAR(255);

> SELECT * FROM customers;
 customer_id |     name      |        email        | age | address
-------------+---------------+---------------------+-----+---------
           1 | Alice Smith   | alice@example.com   |  30 |
           3 | Charlie Brown | charlie@example.com |  35 |
           2 | Bob Johnson   | bob@example.com     |  30 |
           5 | Eve White     | eve@example.com     |  28 |
(4 rows)

INSERT INTO customers (name, email, age, address) VALUES
('Frank Black', 'frank@example.com', 45, '123 Maple Street');

> SELECT * FROM customers;
 customer_id |     name      |        email        | age |     address
-------------+---------------+---------------------+-----+------------------
           1 | Alice Smith   | alice@example.com   |  30 |
           3 | Charlie Brown | charlie@example.com |  35 |
           2 | Bob Johnson   | bob@example.com     |  30 |
           5 | Eve White     | eve@example.com     |  28 |
           6 | Frank Black   | frank@example.com   |  45 | 123 Maple Street
(5 rows)

ALTER TABLE customers DROP COLUMN age;

INSERT INTO customers (name, email, address) VALUES
('Henry Ford', 'henry@example.com', '789 Pine Street');

> select * from customers;
 customer_id |     name      |        email        |     address
-------------+---------------+---------------------+------------------
           1 | Alice Smith   | alice@example.com   |
           3 | Charlie Brown | charlie@example.com |
           2 | Bob Johnson   | bob@example.com     |
           5 | Eve White     | eve@example.com     |
           6 | Frank Black   | frank@example.com   | 123 Maple Street
           7 | Henry Ford    | henry@example.com   | 789 Pine Street
(6 rows)

ALTER TABLE customers RENAME COLUMN name TO full_name;

INSERT INTO customers (full_name, email, address) VALUES ('Ingrid Bergman', 'ingrid@example.com', '321 Birch Lane');